"""
Case Type   : 全链路跟踪
Case Name   : 非管理员用户不可解析全链路信息
Create At   : 2025-10-10
Owner       : sungang14
Description :
    1、开启enable_stmt_track参数gs_guc reload -N all -I all -c "enable_stmt_track=on"
    2、开启全量sql跟踪gs_guc reload -N all -I all -c "track_stmt_stat_level = 'L1,OFF'"
    3、建表并写入数据
    4、当前用户查询trace信息
    5、创建非sysmonitor用户查询trace信息
    6、非sysmonitor用户解析步骤4查询出的trace字段
Expect      :
    1、开启成功
    2、开启成功
    3、操作执行成功
    4、查看trace字段内容不为空
    5、无权限查询
    6、解析失败ERROR: permission denied for relation statement_history
History     :
"""

import os
import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node

logger = Logger()
Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class SlowSQLTestCase(unittest.TestCase):
    def setUp(self):
        logger.info(f'-----{os.path.basename(__file__)} start-----')
        self.constant = Constant()
        self.comsh = CommonSH('PrimaryDbUser')
        self.node = Node("PrimaryDbUser")
        status = self.comsh.get_db_cluster_status('status')
        self.assertTrue(status)
        self.show_enable_stmt = 'show enable_stmt_track;'
        self.set_track_level = "track_stmt_stat_level='L1,OFF'"
        self.set_duration_statement = 'log_min_duration_statement=0'
        self.reset_track_level = "track_stmt_stat_level='OFF,L0'"
        self.reset_duration_statement = 'log_min_duration_statement=30min'
        self.show_sql_count = 'show instr_unique_sql_count;'
        self.set_sql_count = 'instr_unique_sql_count=100'
        self.table_name = 'table_trace_Case0019'
        self.user_name = 'user_case0019'
        self.prepare_data = f"drop table if exists {self.table_name};" \
                            f"CREATE TABLE {self.table_name} (id SERIAL PRIMARY KEY, username VARCHAR(50) NOT NULL, email VARCHAR(100), birth_date DATE, balance NUMERIC(10,2) DEFAULT 0.00);" \
                            f"INSERT INTO {self.table_name} values (generate_series(1, 20),'name1', 'exa.com','2010-09-12', 4.6);" \
                            f"select * from {self.table_name};"
        self.show_track_level = 'show track_stmt_stat_level;'
        self.show_duration_statement = 'show log_min_duration_statement;'
        self.create_user = f"drop user if exists {self.user_name};CREATE USER {self.user_name} PASSWORD '{self.node.db_password}';"
        self.connect_sys = f'-U {self.user_name} -W {self.node.db_password}'
        self.prepare_data2 = f"delete from statement_history where query like '%{self.table_name}%';"
        text = '--清理statement_history表内容--'
        logger.info(text)
        data_cmd = self.comsh.execut_db_sql(self.prepare_data2, dbname='postgres')
        logger.info(data_cmd)
        self.assertIn('DELETE', data_cmd, '执行失败' + text)

    def test_slowsql(self):
        text = '--step1:开启enable_stmt_track参数;expect:开启成功--'
        logger.info(text)
        show_cmd = self.comsh.execut_db_sql(self.show_enable_stmt)
        logger.info(show_cmd)
        self.assertEquals('on', show_cmd.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = "--step2:设置track_stmt_stat_level = 'L1,OFF';expect:设置成功--"
        logger.info(text)
        set_cmd1 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_track_level)
        logger.info(set_cmd1)
        show_res = self.comsh.execut_db_sql(self.show_track_level)
        logger.info(show_res)
        self.assertEquals('L1,OFF', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)

        set_cmd2 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_duration_statement)
        logger.info(set_cmd2)
        show_res = self.comsh.execut_db_sql(self.show_duration_statement)
        logger.info(show_res)
        self.assertEquals('0', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = '--step3:建表并写入数据;expect:成功--'
        logger.info(text)
        data_cmd = self.comsh.execut_db_sql(self.prepare_data, dbname='postgres')
        logger.info(data_cmd)
        self.assertIn('INSERT 0 20', data_cmd, '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, data_cmd, '执行失败' + text)
        self.assertIn(f'(20 rows)', data_cmd, '执行失败' + text)

        text = '--step4:在主机表statement_history中查看trace字段;expect:查看trace字段内容不为空--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query='select * from {self.table_name};';",
            dbname='postgres')
        logger.info(sql_re)
        trace_msg = sql_re.splitlines()[-2].strip()
        logger.info(trace_msg)
        self.assertIsNotNone(trace_msg, '执行失败' + text)

        text = "--step5:创建非sysmonitor用户查询trace信息;expect:无权限查询--"
        logger.info(text)
        data_cmd = self.comsh.execut_db_sql(self.create_user, dbname='postgres')
        logger.info(data_cmd)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, data_cmd, '执行失败' + text)
        permission_denied_msg = f"permission denied for relation statement_history"
        sql_re = self.comsh.execut_db_sql(
            f"select trace from statement_history where query='select * from {self.table_name};';",
            sql_type=self.connect_sys,
            dbname='postgres')
        logger.info(sql_re)
        self.assertIn(permission_denied_msg, sql_re, '执行失败' + text)

        text = "--step6:切换非管理员用户{self.user_name}，在主机反解析步骤5查询出的trace字段;expect:解析失败ERROR: permission denied for relation statement_history--"
        logger.info(text)
        permission_denied_msg = f"permission denied for function statement_trace_decode"
        sql_re4 = self.comsh.execut_db_sql(f"select statement_trace_decode('{trace_msg}');", sql_type=self.connect_sys,
                                           dbname='postgres')
        logger.info(sql_re4)
        self.assertIn(permission_denied_msg, sql_re4, '执行失败' + text)

    def tearDown(self):
        text = '--step7:清理环境，恢复默认值;expect:成功--'
        logger.info(text)
        sql_re = self.comsh.execut_db_sql(f"drop table if exists {self.table_name};", dbname='postgres')
        logger.info(sql_re)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_re, '执行失败' + text)

        sql_re = self.comsh.execut_db_sql(f"drop user if exists {self.user_name};", dbname='postgres')
        logger.info(sql_re)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, sql_re, '执行失败' + text)

        self.comsh.execute_gsguc('reload',
                                 self.constant.GSGUC_SUCCESS_MSG,
                                 self.reset_track_level)
        show_res = self.comsh.execut_db_sql(self.show_track_level)
        logger.info(show_res)
        self.assertEquals('OFF,L0', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        self.comsh.execute_gsguc('reload',
                                 self.constant.GSGUC_SUCCESS_MSG,
                                 self.reset_duration_statement)
        show_res = self.comsh.execut_db_sql(self.show_duration_statement)
        logger.info(show_res)
        self.assertEquals('30min', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        set_cmd1 = self.comsh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            self.set_sql_count)
        logger.info(set_cmd1)
        show_res = self.comsh.execut_db_sql(self.show_sql_count)
        logger.info(show_res)
        self.assertEquals('100', show_res.splitlines()[-2].strip(),
                          '执行失败' + text)
        logger.info(f'-----{os.path.basename(__file__)} end-----')
